package com.example.grpc.proxy;

import io.grpc.stub.ClientCalls;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http2.*;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

@Slf4j
@ChannelHandler.Sharable
public class CustHttp2ServerHandler extends Http2ChannelDuplexHandler {

    private static final String HOST = "127.0.0.1";
    private static final int PORT = 10511;

    private Http2HeadersFrame http2HeadersFrame = null;
    private final BlockingQueue<Object> responseQueue;

    public CustHttp2ServerHandler() {
        this.responseQueue = new ArrayBlockingQueue<>(1);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof Http2StreamFrame) {
            Http2StreamFrame streamFrame = (Http2StreamFrame) msg;
            log.info("server stream id:{}, state: {}", streamFrame.stream().id(), streamFrame.stream().state());
        }
        if (msg instanceof Http2HeadersFrame) {
            Http2HeadersFrame headersFrame = (Http2HeadersFrame) msg;
            http2HeadersFrame = headersFrame;
            log.info("server header received:{}", headersFrame.headers());
        } else if (msg instanceof Http2SettingsFrame) {
            Http2SettingsFrame settingsFrame = (Http2SettingsFrame) msg;
            log.info("server settings:{}", settingsFrame.settings());
        } else if (msg instanceof Http2SettingsAckFrame) {
            Http2SettingsAckFrame ackFrame = (Http2SettingsAckFrame) msg;
            log.info("server ack frame:{}", ackFrame.name());
        } else if (msg instanceof Http2DataFrame) {
            Http2DataFrame dataFrame = (Http2DataFrame) msg;
            log.info("server data received:");
            final EventLoopGroup clientWorkerGroup = new NioEventLoopGroup();
            final Bootstrap b = new Bootstrap();
            b.group(clientWorkerGroup);
            b.channel(NioSocketChannel.class);
            b.remoteAddress(HOST, PORT);
            b.handler(new CustHttp2ClientInit(http2HeadersFrame, dataFrame, ctx));
            //b.handler(new Http2ClientFrameInitializer(http2HeadersFrame, dataFrame, responseQueue));

            // 启动客户端
            log.info("[" + HOST + ':' + PORT + ']');
            Channel channel = b.connect().syncUninterruptibly().channel();

            Http2StreamChannelBootstrap streamChannelBootstrap = new Http2StreamChannelBootstrap(channel);
            Http2StreamChannel http2StreamChannel = streamChannelBootstrap.open().syncUninterruptibly().get();
            http2HeadersFrame.stream(http2StreamChannel.stream());
            http2StreamChannel.pipeline().addLast(new Http2OutboundClientHandler());
            responseQueue.take();
            channel.close();
            log.info("server end.........");
        } else {
            log.error("server other msg");
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        log.info("********************* channelReadComplete *******************************");
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("exception--------------------------------------------");
        cause.printStackTrace();
        ctx.close();
    }

}
